package com.flink.demo.redis;

import com.kingstar.flink.process.AbstractAsyncProcess;
import com.kingstar.flink.type.DataRow;
import com.kingstar.flink.type.RowSchema;
import com.kingstar.stream.table.BaseTable;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.types.Row;

/**
 * @author xin.ding
 * @date 2021/3/18 10:52
 */
public class JoinRedisFunction extends AbstractAsyncProcess<Row, DataRow> {

    public JoinRedisFunction(BaseTable asyncTable) {
        super(asyncTable);
    }

    @Override
    public DataRow process(Row input) {

        String key = (String) input.getField(0);

        final Object value = getCache(key);
        DataRow dataRow = new DataRow(2);
        dataRow.setField(0, key);
        dataRow.setField(1, value);

        return dataRow;

    }

    @Override
    public RowSchema setResultType() {
        return RowSchema.build(new String[][]{
                {"c1", "string"},
                {"c2", "string"}
        });
    }

    @Override
    public void configure(Configuration runProperties) {
    }
}
